package com.bff.gaia.mix.api.xjoin2.operators;

import com.bff.gaia.api.java.tuple.Tuple5;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.mix.api.datastream.DataMixStream;
import com.bff.gaia.mix.api.datastream.SingleOutputMixStreamOperator;
import com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment;
import com.bff.gaia.mix.api.functions.source.RichSourceFunction;
import com.bff.gaia.mix.api.transformations.MixStreamTransformation;
import com.bff.gaia.mix.api.transformations.TwoInputTransformation;
import com.bff.gaia.mix.api.xjoin2.client.mysql.all.MysqlAllSyncXjoinFunction;
import com.bff.gaia.mix.api.xjoin2.client.mysql.all.MysqlAllXjoinSourceFunction;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import com.bff.gaia.mix.api.xjoin2.functions.SyncXjoinFunction;
import com.bff.gaia.mix.api.xjoin2.partition.SideKeySclector;
import com.bff.gaia.mix.api.xjoin2.partition.StreamKeySelector;
import com.bff.gaia.mix.api.xjoin2.partition.XjoinPartitioner;
import com.bff.gaia.types.Row;

import java.util.List;

public class SyncXjoinOperatorGateWay {
	private static SyncXjoinFunction<Row> loadAllXjoinFunction(
		StreamTableInfo streamTableInfo,
		SideTableInfo sideTableInfo,
		JoinInfo joinInfo,
		List<FieldInfo> outFieldInfoList) {
		String sideType = sideTableInfo.getType();
		switch (sideType.toLowerCase()) {
			case "mysql": {
				return new MysqlAllSyncXjoinFunction(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
			}
			default: {
				return null;
			}
		}
	}

	private static RichSourceFunction<Tuple5<Integer, String, String, String, Integer>> loadAllXjoinSourceFunction(
		StreamTableInfo streamTableInfo,
		SideTableInfo sideTableInfo,
		JoinInfo joinInfo,
		List<FieldInfo> outFieldInfoList) {
		String sideType = sideTableInfo.getType();
		switch (sideType.toLowerCase()) {
			case "mysql": {
				return new MysqlAllXjoinSourceFunction(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
			}
			default: {
				return null;
			}
		}
	}

	private static SingleOutputMixStreamOperator<Tuple5<Integer, String, String, String, Integer>> loadXjoinSource(
		MixStreamExecutionEnvironment environment,
		StreamTableInfo streamTableInfo,
		SideTableInfo sideTableInfo,
		JoinInfo joinInfo,
		List<FieldInfo> outFieldInfoList
	) {
		RichSourceFunction<Tuple5<Integer, String, String, String, Integer>> xjoinSourceFunction = loadAllXjoinSourceFunction(streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);
		return environment.addSource(xjoinSourceFunction);
	}

	public static SingleOutputMixStreamOperator<Row> getSyncXjoinStream(MixStreamExecutionEnvironment environment,
	                                                                    DataMixStream<Row> stream,
	                                                                    StreamTableInfo streamTableInfo,
	                                                                    SideTableInfo sideTableInfo,
	                                                                    JoinInfo joinInfo,
	                                                                    List<FieldInfo> outFieldInfoList,
	                                                                    RowTypeInfo outTypeInfo,
	                                                                    String shareSlot,
	                                                                    int... streamTableKeyPos
	) {
		SyncXjoinFunction<Row> syncXjoinFunction = loadAllXjoinFunction(
			streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);

		SingleOutputMixStreamOperator<Tuple5<Integer, String, String, String, Integer>> sideTableSource = loadXjoinSource(
			environment, streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList);



		DataMixStream<Row> partitionStream;
		DataMixStream<Tuple5<Integer, String, String, String, Integer>> partitionSideTableSource;

		if (joinInfo.isPartitionedJoin()) {
			partitionSideTableSource = sideTableSource
				.partitionCustom(new XjoinPartitioner(), new SideKeySclector());
		} else {
			partitionSideTableSource = sideTableSource
				.broadcast();
		}


		partitionStream = stream.partitionCustom(new XjoinPartitioner(), new StreamKeySelector(joinInfo.getParallelism(), streamTableKeyPos));


		TwoInputTransformation<Row, Tuple5<Integer, String, String, String, Integer>, Row> transform ;
		if (shareSlot.equals("")){
			transform = new TwoInputTransformation<>(
				partitionStream.getTransformation(),
				partitionSideTableSource.getTransformation(),
				"Sync-XJoin",
				new SyncXjoinOperator<>(syncXjoinFunction, streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList),
				outTypeInfo,
				joinInfo.getParallelism());
		}else {
			MixStreamTransformation<Tuple5<Integer, String, String, String, Integer>> t =partitionSideTableSource.getTransformation();
			t.setSlotSharingGroup(shareSlot);
			transform = new TwoInputTransformation<>(
				partitionStream.getTransformation(),
				t,
				"Sync-XJoin",
				new SyncXjoinOperator<>(syncXjoinFunction, streamTableInfo, sideTableInfo, joinInfo, outFieldInfoList),
				outTypeInfo,
				joinInfo.getParallelism());
		}

		transform.setParallelism(joinInfo.getParallelism());
		//TODO make SingleOutputMixStreamOperator protect
		SingleOutputMixStreamOperator<Row> returnMixStream = new SingleOutputMixStreamOperator<Row>(environment, transform);
		environment.addOperator(transform);
		return returnMixStream;
	}
}